package com.aotain.nyx.flink;


import java.io.FileInputStream;
import java.io.InputStream;
import java.util.HashMap;
import java.util.Properties;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.SplitStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer08;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;

import com.aotain.common.ZkKafkaConfig;
import com.aotain.nyx.abnormal.AbnormalCheckImport;
import com.aotain.nyx.abnormal.AbnormalReduce;
import com.aotain.nyx.abnormal.AbnormalReverseCheckImport;
import com.aotain.nyx.abnormal.AbnormalReverseReduce;
import com.aotain.nyx.abnormal.AbnormalReverseStatMap;
import com.aotain.nyx.abnormal.AbnormalStatMap;
import com.aotain.nyx.abnormal.AbnormalStatTuple;
import com.aotain.nyx.ackflood.AckFloodFlink;
import com.aotain.nyx.ackflood.AckFloodReduce;
import com.aotain.nyx.ackflood.AckFloodStatMap;
import com.aotain.nyx.ackflood.AckFloodTuple;
import com.aotain.nyx.allattstatd.AllAttDStatisImport;
import com.aotain.nyx.allattstatd.AllAttStatMap;
import com.aotain.nyx.allattstatd.AllAttStatisReduce;
import com.aotain.nyx.appstatd.AppAttDStatisImport;
import com.aotain.nyx.appstatd.AppAttStatMap;
import com.aotain.nyx.appstatd.AppAttStatisReduce;
import com.aotain.nyx.areastatis.AreaStatMap;
import com.aotain.nyx.areastatis.AreaStatisImport;
import com.aotain.nyx.areastatis.AreaStatisReduce;
import com.aotain.nyx.bots.BotsFlink;
import com.aotain.nyx.bots.BotsReduce;
import com.aotain.nyx.bots.BotsStatMap;
import com.aotain.nyx.bots.BotsTuple;
import com.aotain.nyx.cc.CCDetectFlink;
import com.aotain.nyx.cc.URLStatMap;
import com.aotain.nyx.cc.URLStatReduce;
import com.aotain.nyx.commandinjection.CommandInjectionPost;
import com.aotain.nyx.common.PostTuple;
import com.aotain.nyx.common.URLStatTuple;
import com.aotain.nyx.dnsflood.DnsFloodFixedImport;
import com.aotain.nyx.dnsflood.DnsFloodFixedStatMap;
import com.aotain.nyx.dnsflood.DnsFloodForgedImport;
import com.aotain.nyx.dnsflood.DnsFloodForgedStatMap;
import com.aotain.nyx.dnsflood.DnsFloodReduce;
import com.aotain.nyx.dnsflood.DnsFloodTuple;
import com.aotain.nyx.flowstatd.FlowAttDStatisImport;
import com.aotain.nyx.flowstatd.FlowAttStatMap;
import com.aotain.nyx.flowstatd.FlowAttStatisReduce;
import com.aotain.nyx.icmpflood.ICMPFloodDetect;
import com.aotain.nyx.icmpflood.ICMPFloodMap;
import com.aotain.nyx.icmpflood.ICMPFloodReduce;
import com.aotain.nyx.ipstat.IPFlowStatImport;
import com.aotain.nyx.ipstat.IPFlowStatMap;
import com.aotain.nyx.ipstat.IPFlowStatReduce;
import com.aotain.nyx.ipstat.IPStatTuple;
import com.aotain.nyx.port.PortAbnReduce;
import com.aotain.nyx.port.PortAbnStatis;
import com.aotain.nyx.port.PortAbnStatisMap;
import com.aotain.nyx.port.PortDetect;
import com.aotain.nyx.port.PortMap;
import com.aotain.nyx.port.PortReduce;
import com.aotain.nyx.postdetecter.PostDetecter;
import com.aotain.nyx.scanattack.ScanAttackFlink;
import com.aotain.nyx.scanattack.ScanAttackReduce;
import com.aotain.nyx.scanattack.ScanAttackStatMap;
import com.aotain.nyx.scanattack.ScanAttackTuple;
import com.aotain.nyx.sensitiveurl.SensitiveUrlDetect;
import com.aotain.nyx.sourstatd.SourDStatisImport;
import com.aotain.nyx.sourstatd.SourDStatisReduce;
import com.aotain.nyx.sourstatd.SourStatDMap;
import com.aotain.nyx.sqlinjection.SQLInjectionDetect;
import com.aotain.nyx.sqlinjection.SQLInjectionMap;
import com.aotain.nyx.sqlinjection.SQLInjectionReduce;
import com.aotain.nyx.sqlinjection.SQLInjectionTuple;
import com.aotain.nyx.statis.AbnStatisMap;
import com.aotain.nyx.statis.AbnStatisTuple;
import com.aotain.nyx.synflood.SynfloodFixedImport;
import com.aotain.nyx.synflood.SynfloodFixedStatMap;
import com.aotain.nyx.synflood.SynfloodForgedImport;
import com.aotain.nyx.synflood.SynfloodForgedStatMap;
import com.aotain.nyx.synflood.SynfloodReduce;
import com.aotain.nyx.synflood.SynfloodTuple;
import com.aotain.nyx.udpflood.UDPFloodDetect;
import com.aotain.nyx.udpflood.UDPFloodMap;
import com.aotain.nyx.udpflood.UDPFloodReduce;
import com.aotain.nyx.util.CommonUtil;
import com.aotain.nyx.util.FilterNullEnum;
import com.aotain.nyx.util.FilterNullObject;
import com.aotain.nyx.vc.VCDetect;
import com.aotain.nyx.vc.VCMap;
import com.aotain.nyx.vc.VCReduce;
import com.aotain.nyx.xss.XSSDetect;
import com.aotain.nyx.xss.XSSMap;
import com.aotain.nyx.xss.XSSReduce;
import com.aotain.nyx.xss.XSSTuple;

/**
 * Flink Nyx 项目主类
 * Start Cmd：
 * /usr/local/flink/bin/flink run -c com.aotain.nyx.flink.NyxFlinkStart 
 * 		/home/hadoop/nyx/lib/hades-nyx-0.0.1-SNAPSHOT.jar
 * @author Administrator
 *
 */
public class NyxFlinkStart {

	public static void main(String[] args) throws Exception {

//		String dbJsonSDS = "{\"driverClassName\":\"com.mysql.jdbc.Driver\","
//				+ "\"maxActive\":\"100\",\"maxIdle\":\"25\",\"maxWait\":\"10000\","
//				+ "\"name\":\"sds\",\"password\":\"h16aug8v3w\",\"queryTimeout\":\"60\","
//				+ "\"type\":\"javax.sql.DataSource\",\"url\":\"jdbc:mysql://192.168.5.28:3306/SDS\""
//				+ ",\"userName\":\"sds\""
//				+ ",\"validationQuery\":\"SELECT SYSDATE()\"}";
//		String dbJsonNYX = "{\"driverClassName\":\"com.mysql.jdbc.Driver\","
//				+ "\"maxActive\":\"100\",\"maxIdle\":\"25\",\"maxWait\":\"10000\","
//				+ "\"name\":\"sds\",\"password\":\"h16aug8v3w\",\"queryTimeout\":\"60\","
//				+ "\"type\":\"javax.sql.DataSource\",\"url\":\"jdbc:mysql://192.168.5.28:3306/NYX\","
//				+ "\"userName\":\"sds\",\"validationQuery\":\"SELECT SYSDATE()\"}";
		Properties prop = new Properties();
		InputStream in = new FileInputStream("/home/storm/config/dbconfig.ini");
		prop.load(in);
		String dbJsonSDS = prop.getProperty("dbJsonSDS");
		String dbJsonNYX = prop.getProperty("dbJsonNYX");
		String filterIpSql = "SELECT DISTINCT A.IP FROM SDS_SERVERINFO A JOIN SDS_WEBSITEINFO B ON A.SITEID = B.SITEID WHERE B.MONITORSTATUS = 1";
		String whiteIpSql = " SELECT * FROM SDS_WHITELIST";
		
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		//设置事件时间  2022-04-30 turk
		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
		ZkKafkaConfig z_kConfig = new ZkKafkaConfig("/home/storm/config/dbconfig.ini");

		Properties properties = new Properties();
//		properties.setProperty("bootstrap.servers", CommonUtil.KAFKA_BROKER);
		properties.setProperty("bootstrap.servers", z_kConfig.getBrokerList());
		// only required for Kafka 0.8
//		properties.setProperty("zookeeper.connect", CommonUtil.ZOOKEEPER_HOST);
		properties.setProperty("zookeeper.connect", z_kConfig.getZkConn_());
		properties.setProperty("group.id", CommonUtil.GROUP_ID);
//		properties.setProperty("group.id", "flink");
		
		Properties propertiesPost = new Properties();
//		propertiesPost.setProperty("bootstrap.servers", CommonUtil.KAFKA_BROKER);
		propertiesPost.setProperty("bootstrap.servers", z_kConfig.getBrokerList());
		// only required for Kafka 0.8
//		propertiesPost.setProperty("zookeeper.connect", CommonUtil.ZOOKEEPER_HOST);
		propertiesPost.setProperty("zookeeper.connect", z_kConfig.getZkConn_());
		propertiesPost.setProperty("group.id", CommonUtil.GROUP_ID_POST);
		
		//ZkKafkaConfig z_kConfig = new ZkKafkaConfig("D:\\workspace\\Hades-Z\\hades-main\\hades-project\\config\\dbconfig.ini");


//		DataStream<Tuple2<String,String>> dataStream = env.addSource(new FlinkKafkaConsumer08<>(CommonUtil.TOPIC, new SimpleStringSchema(), properties))
//				.map(new SourceDataClean(dbJson)).keyBy(0);

		//清洗后的数据流，包含了IP归属地信息
		//		DataStream<Tuple2<String, String>> dataStream = env.socketTextStream("192.168.5.96", 9988)
		//				.map(new SourceDataMap())
		//				.map(new SourceDataClean()).keyBy(0);
//		DataStream<Tuple2<String, String>> sourceStream = env.socketTextStream("192.168.5.96", 9988)
//				.map(new SourceDataFilterMap(filterIpSql,whiteIpSql, dbJson_oracle))
//				.filter(new FilterNullObject<String>(FilterNullEnum.F1ISNOTNULL));

		DataStream<Tuple2<String,String>> sourceStream = null;


		//分流输出
		if(z_kConfig.getFlinkMode().toUpperCase().equals("KAFKA")) {
			sourceStream = env.addSource(new FlinkKafkaConsumer08<>(CommonUtil.TOPIC, new SimpleStringSchema(), properties))
					.map(new SourceDataFilterMap(filterIpSql,whiteIpSql, dbJsonSDS, dbJsonNYX))
					.filter(new FilterNullObject<String>(FilterNullEnum.F1ISNOTNULL)).name("UDII-SOURCE");
		} else if (z_kConfig.getFlinkMode().toUpperCase().equals("SOCKET")) {
			//多个端口监听
			String port = z_kConfig.getFlinkSocketPort();
			for(String p : port.split("\\|")) {
				if(sourceStream == null) {
					sourceStream = env.socketTextStream(z_kConfig.getFlinkSocketIP(), Integer.parseInt(p))
							.map(new SourceDataFilterMap(filterIpSql,whiteIpSql, dbJsonSDS, dbJsonNYX))
							.filter(new FilterNullObject<String>(FilterNullEnum.F1ISNOTNULL))
							.name("UDII-SOURCE-" + p);
					
				} else {
					sourceStream = sourceStream.union(env.socketTextStream(z_kConfig.getFlinkSocketIP(), Integer.parseInt(p))
							.map(new SourceDataFilterMap(filterIpSql,whiteIpSql, dbJsonSDS, dbJsonNYX))
							.filter(new FilterNullObject<String>(FilterNullEnum.F1ISNOTNULL))
							.name("UDII-SOURCE-" + p));
				}
			}
			
			if(sourceStream == null) {
				System.out.println("sourceStream ERROR");
				return;
			}
		} else {
			System.out.println("Config Flink Mode is ERROR");
			return;
		}


		//设置时间字段的水位线
		DataStream<Tuple2<String, String>> streamResult= sourceStream.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<Tuple2<String, String>>() {
			@Override
			public long extractAscendingTimestamp(Tuple2<String, String> element) {
				String[] str = element.f1.split("|");
				return Long.parseLong(str[21]);
			}
		});
		
		/*2017-07-25 新增Post 数据接入 by turk*/
//		DataStream<Tuple2<String,PostTuple>> sourceStreamPost = env.addSource(new FlinkKafkaConsumer08<>(CommonUtil.TOPIC_POST, new SimpleStringSchema(), propertiesPost))
//				.map(new SourceDataPost(filterIpSql,whiteIpSql, dbJsonSDS, dbJsonNYX))
//				.filter(new FilterNullObject<PostTuple>(FilterNullEnum.F1ISNOTNULL)).name("POST-SOURCE");
		
		SplitStream<Tuple2<String,String>> splitStream = streamResult.split(new SourceDataSelector());



		DataStream<Tuple2<String, String>> dipStream = splitStream.select("DipStream")
				.map(new DipSoucreStreamStat());
		
		DataStream<Tuple2<String, String>> sipStream = splitStream.select("SipStream")
				.map(new SipSoucreStreamStat());

		//** IP 流量统计
		DataStream<Tuple2<String, IPStatTuple>> ipStatStream = dipStream.map(new IPFlowStatMap())
					.name("IPFlowStatMap-MAP"); 

		SingleOutputStreamOperator<MongoSinkObject>  
			time5Min = ipStatStream.keyBy(0)
						.window(EventTimeSessionWindows.withGap(Time.seconds(60))).allowedLateness(Time.seconds(60))
							       .reduce(new IPFlowStatReduce())
							       .flatMap(new IPFlowStatImport("SDS_SESSION_STAT_5MIN")).name("IPSTAT-5MIN");

		SingleOutputStreamOperator<MongoSinkObject>  
			time30Min = ipStatStream.keyBy(0)
						.window(EventTimeSessionWindows.withGap(Time.minutes(5))).allowedLateness(Time.seconds(60))
									.reduce(new IPFlowStatReduce())
									.flatMap(new IPFlowStatImport("SDS_SESSION_STAT_30MIN")).name("IPSTAT-30MIN");

		SingleOutputStreamOperator<MongoSinkObject>  
			time60Min = ipStatStream.keyBy(0)
				.window(EventTimeSessionWindows.withGap(Time.minutes(5))).allowedLateness(Time.seconds(60))
				                    .reduce(new IPFlowStatReduce())
				                    .flatMap(new IPFlowStatImport("SDS_SESSION_STAT_HOUR")).name("IPSTAT-HOUR");
		
		SingleOutputStreamOperator<MongoSinkObject>   
			timeDay = ipStatStream.keyBy(0)
				.window(EventTimeSessionWindows.withGap(Time.minutes(5))).allowedLateness(Time.seconds(60))
								  .reduce(new IPFlowStatReduce())
								  .flatMap(new IPFlowStatImport("SDS_SESSION_STAT_DAY")).name("IPSTAT-DAY");

		//**ICMP Flood
		SingleOutputStreamOperator<AbnStatisTuple>  
			winIcmp = dipStream.map(new ICMPFloodMap())
					           .filter(new FilterNullObject<BasicTuple>(FilterNullEnum.F1ISNOTNULL))
				               .keyBy(0)
				.window(EventTimeSessionWindows.withGap(Time.minutes(1))).allowedLateness(Time.seconds(60))
				               .reduce(new ICMPFloodReduce())
				               .flatMap(new ICMPFloodDetect()).name("ICMP Flood");
		
		//**CC 攻击检测
		SingleOutputStreamOperator<AbnStatisTuple>  
			winCC = dipStream.map(new URLStatMap())
				           .filter(new FilterNullObject<URLStatTuple>(FilterNullEnum.F1ISNOTNULL))
			               .keyBy(0)
				.window(EventTimeSessionWindows.withGap(Time.minutes(1))).allowedLateness(Time.seconds(60))
			               .reduce(new URLStatReduce())
			               .flatMap(new CCDetectFlink()).name("CC");
		//**端口异常检测
		SingleOutputStreamOperator<AbnStatisTuple>  
			 winPort =	dipStream.map(new PortMap(dbJsonSDS))
				       .filter(new FilterNullObject<BasicTuple>(FilterNullEnum.F1ISNOTNULL))
		               .keyBy(0)
						.window(EventTimeSessionWindows.withGap(Time.minutes(1))).allowedLateness(Time.seconds(60))
				       .reduce(new PortReduce())
				       .flatMap(new PortDetect(dbJsonSDS)).name("Port Detect");
		
		winPort.map(new PortAbnStatisMap()).name("Port ABN MAP")
				.keyBy(0)
				.window(EventTimeSessionWindows.withGap(Time.minutes(1))).allowedLateness(Time.seconds(60))
				.reduce(new PortAbnReduce())
				.flatMap(new PortAbnStatis()).name("Port ABN Statis");

		
		
		
		SingleOutputStreamOperator<AbnStatisTuple>  
			 winSQLInjection = dipStream.map(new SQLInjectionMap(z_kConfig.getSQLInjection()))
					       .filter(new FilterNullObject<SQLInjectionTuple>(FilterNullEnum.F1ISNOTNULL))
			               .keyBy(0)
				.window(EventTimeSessionWindows.withGap(Time.minutes(1))).allowedLateness(Time.seconds(60))
			               .reduce(new SQLInjectionReduce())
			               .flatMap(new SQLInjectionDetect()).name("SQL Injection");

//		/*POST SQL注入*/
//		SingleOutputStreamOperator<AbnStatisTuple>  
//			winSQLInjectionPost = sourceStreamPost
//		               .keyBy(0)
//		               .flatMap(new InjectionDetectPost(z_kConfig.getSQLInjection())).name("POST SQL Injection");
//		/*POST XSS*/
//	    SingleOutputStreamOperator<AbnStatisTuple>  
//			winXSSPost = sourceStreamPost
//	               .keyBy(0)
//	               .flatMap(new XSSDetectPost()).name("XSS POST");
	    
	    /*POST Detecter*/
//	    SingleOutputStreamOperator<Tuple2<String, String>>  
//			winPostDetecter = sourceStreamPost
//	               .keyBy(0)
//	               .map(new PostDetecter()).name("POST Detecter");
//	    
//	    
//	    /*POST COMMAND注入*/
//	    SingleOutputStreamOperator<AbnStatisTuple>  
//	    	winCommandInjectionPost = sourceStreamPost
//				    .keyBy(0)
//				    .flatMap(new CommandInjectionPost()).name("COMMAND INJECTION POST");
	    
	    /*Sensitive URL*/
	    SingleOutputStreamOperator<AbnStatisTuple>  
			winSensitiveUrl = dipStream.map(new URLStatMap())
			           .filter(new FilterNullObject<URLStatTuple>(FilterNullEnum.F1ISNOTNULL))
		               .keyBy(0)
		               .flatMap(new SensitiveUrlDetect()).name("Sensitive Url");
		
		SingleOutputStreamOperator<AbnStatisTuple>  
			winXSS = dipStream.map(new XSSMap())
				       .filter(new FilterNullObject<XSSTuple>(FilterNullEnum.F1ISNOTNULL))
		               .keyBy(0)
		               .timeWindow(Time.minutes(1))
		               .reduce(new XSSReduce())
		               .flatMap(new XSSDetect()).name("XSS Detect");
		
		SingleOutputStreamOperator<AbnStatisTuple>  
			winVC = dipStream.map(new VCMap())
				       .filter(new FilterNullObject<BasicTuple>(FilterNullEnum.F1ISNOTNULL))
		               .keyBy(0)
		               .timeWindow(Time.minutes(1))
		               .reduce(new VCReduce())
		               .flatMap(new VCDetect()).name("VC Detect");
	
		//***SynFlood检测
		//---固定源
		SingleOutputStreamOperator<AbnStatisTuple>
			winSynFixed = dipStream.map(new SynfloodFixedStatMap())
											.filter(new FilterNullObject<SynfloodTuple>(FilterNullEnum.F0ISNOTNULL))
											.keyBy(0)
											.timeWindow(Time.minutes(1))
											.reduce(new SynfloodReduce())
											.flatMap(new SynfloodFixedImport()).name("SYN Flood");
		//---伪造源
		SingleOutputStreamOperator<AbnStatisTuple>
			winSynForged = dipStream.map(new SynfloodForgedStatMap())
										.filter(new FilterNullObject<SynfloodTuple>(FilterNullEnum.F0ISNOTNULL))
										.keyBy(0)
										.window(EventTimeSessionWindows.withGap(Time.minutes(1))).allowedLateness(Time.seconds(60))
										.reduce(new SynfloodReduce())
										.flatMap(new SynfloodForgedImport()).name("SYN Flood Forged");
		//--UDP Flood
		SingleOutputStreamOperator<AbnStatisTuple>
			winUdpFlood = dipStream.map(new UDPFloodMap())
								 .filter(new FilterNullObject<HashMap<String,BasicTuple>>(FilterNullEnum.F0ISNOTNULL))
								 .keyBy(0)
				                 .window(EventTimeSessionWindows.withGap(Time.minutes(1))).allowedLateness(Time.seconds(60))
								 .reduce(new UDPFloodReduce())
								 .flatMap(new UDPFloodDetect()).name("UDP Flood");
		
		
		//---异常流量
		SingleOutputStreamOperator<AbnStatisTuple>
			winAbn = dipStream.map(new AbnormalStatMap())
				   .filter(new FilterNullObject<AbnormalStatTuple>(FilterNullEnum.F1ISNOTNULL))
	               .keyBy(0)
				.window(EventTimeSessionWindows.withGap(Time.minutes(1))).allowedLateness(Time.seconds(60))
	               .reduce(new AbnormalReduce())
	               .flatMap(new AbnormalCheckImport()).name("AbnormalCheck");
		
//		SingleOutputStreamOperator<AbnStatisTuple>
//			winAbnRe = sipStream.map(new AbnormalReverseStatMap())
//				   .filter(new FilterNullObject<AbnormalStatTuple>(FilterNullEnum.F1ISNOTNULL))
//	               .keyBy(0)
//	               .timeWindow(Time.minutes(1))
//	               .reduce(new AbnormalReverseReduce())
//	               .flatMap(new AbnormalReverseCheckImport()).name("AbnormalReverseCheck");
		
		//---僵尸主机
		SingleOutputStreamOperator<AbnStatisTuple>
			winBots = dipStream.map(new BotsStatMap())
				   .filter(new FilterNullObject<BotsTuple>(FilterNullEnum.F1ISNOTNULL))
	               .keyBy(0)
	               .timeWindow(Time.minutes(1))
	               .reduce(new BotsReduce())
	               .flatMap(new BotsFlink()).name("Bots");
		
		//---扫描攻击
		SingleOutputStreamOperator<AbnStatisTuple>
			winScan = dipStream.map(new ScanAttackStatMap())
				   .filter(new FilterNullObject<ScanAttackTuple>(FilterNullEnum.F1ISNOTNULL))
	               .keyBy(0)
	               .timeWindow(Time.minutes(1))
	               .reduce(new ScanAttackReduce())
	               .flatMap(new ScanAttackFlink()).name("Scan Attack");
		
		//---AckFlood
		SingleOutputStreamOperator<AbnStatisTuple>
			winAck = dipStream.map(new AckFloodStatMap())
				   .filter(new FilterNullObject<AckFloodTuple>(FilterNullEnum.F1ISNOTNULL))
	               .keyBy(0)
	               .timeWindow(Time.minutes(1))
	               .reduce(new AckFloodReduce())
	               .flatMap(new AckFloodFlink()).name("Ack Flood");
		
		//***DNSFlood
		//---固定源
		SingleOutputStreamOperator<AbnStatisTuple>
			winDnsFixed = dipStream.map(new DnsFloodFixedStatMap())
					.filter(new FilterNullObject<DnsFloodTuple>(FilterNullEnum.F1ISNOTNULL))
					.keyBy(0)
					.timeWindow(Time.minutes(1))
					.reduce(new DnsFloodReduce())
					.flatMap(new DnsFloodFixedImport()).name("DNS Flood");
		//---伪造源
		SingleOutputStreamOperator<AbnStatisTuple>
			winDnsForged = dipStream.map(new DnsFloodForgedStatMap())
					.filter(new FilterNullObject<DnsFloodTuple>(FilterNullEnum.F1ISNOTNULL))
					.keyBy(0)
					.timeWindow(Time.minutes(1))
					.reduce(new DnsFloodReduce())
					.flatMap(new DnsFloodForgedImport()).name("DNS Flood Forged");
		
//		winSynFixed.map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(3))
//			.reduce(new IPFlowStatReduce()).flatMap(new IPFlowStatImport("SDS_SESSION_STAT_5MIN"));
//		winSynFixed.map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(5))
//			.reduce(new IPFlowStatReduce()).flatMap(new IPFlowStatImport("SDS_SESSION_STAT_30MIN"));
//		winSynFixed.map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(5))
//			.reduce(new IPFlowStatReduce()).flatMap(new IPFlowStatImport("SDS_SESSION_STAT_HOUR"));

		//--异常统计---
		winCC.union(winPort)
			.union(winIcmp)
			.union(winSQLInjection)
			.union(winXSS)
			.union(winVC)
			.union(winSynFixed)
			.union(winSynForged)
			.union(winScan)
			.union(winAck)
			.union(winBots)
			.union(winAbn)
			.union(winUdpFlood)
			.map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(3))
			.reduce(new IPFlowStatReduce()).flatMap(new IPFlowStatImport("SDS_SESSION_STAT_5MIN"));
		
		winCC.union(winPort)
			 .union(winIcmp)
			 .union(winSQLInjection)
			 .union(winXSS)
			 .union(winVC)
			 .union(winSynFixed)
			 .union(winSynForged)
			 .union(winScan)
			 .union(winAck)
			 .union(winBots)
			 .union(winAbn)
			 .union(winUdpFlood)
			 .map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(5))
			 .reduce(new IPFlowStatReduce()).flatMap(new IPFlowStatImport("SDS_SESSION_STAT_30MIN"));
		
		winCC.union(winPort)
			 .union(winIcmp)
			 .union(winSQLInjection)
			 .union(winXSS)
			 .union(winVC)
			 .union(winSynFixed)
			 .union(winSynForged)
			 .union(winScan)
			 .union(winAck)
			 .union(winBots)
			 .union(winAbn)
			 .union(winUdpFlood)
			 .map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(5))
			 .reduce(new IPFlowStatReduce()).flatMap(new IPFlowStatImport("SDS_SESSION_STAT_HOUR"));
		
		
		winCC.union(winPort)
			 .union(winIcmp)
			 .union(winSQLInjection)
			 .union(winXSS)
			 .union(winVC)
			 .union(winSynFixed)
			 .union(winSynForged)
			 .union(winScan)
			 .union(winAck)
			 .union(winBots)
			 .union(winAbn).map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(5))
			 .reduce(new IPFlowStatReduce()).flatMap(new IPFlowStatImport("SDS_SESSION_STAT_DAY"));
		
		//SDS_ABNORMAL_AREA表
		winCC.union(winPort)
			 .union(winIcmp)
			 .union(winSQLInjection)
			 .union(winXSS)
			 .union(winVC)
			 .union(winSynFixed)
			 .union(winSynForged)
			 .union(winScan)
			 .union(winAck)
			 .union(winBots)
			 .union(winAbn)
//			 .union(winAbnRe)
			 .union(winUdpFlood)
			 .map(new AreaStatMap()).keyBy(0).timeWindow(Time.minutes(5))
			 .reduce(new AreaStatisReduce()).flatMap(new AreaStatisImport()).name("SDS_ABNORMAL_AREA");
		
		//SDS_ABN_SESS_SOUR_STAT_D
		winCC.union(winPort)
			.union(winIcmp)
			.union(winSQLInjection)
			.union(winXSS)
			.union(winVC)
			.union(winSynFixed)
			.union(winSynForged)
			.union(winScan)
			.union(winAck)
			.union(winBots)
			.union(winAbn)
//			.union(winAbnRe)
			.union(winUdpFlood)
			.map(new SourStatDMap()).keyBy(0).timeWindow(Time.minutes(5))
			.reduce(new SourDStatisReduce()).flatMap(new SourDStatisImport()).name("SDS_ABN_SESS_SOUR_STAT_D");
		
		//应用攻击统计天表SDS_APP_ATT_D
		winCC.union(winPort)
			.union(winIcmp)
			.union(winSQLInjection)
			.union(winXSS)
			.union(winVC)
			.union(winSynFixed)
			.union(winSynForged)
			.union(winScan)
			.union(winAck)
			.union(winBots)
			.union(winAbn)
//			.union(winAbnRe)
			.map(new AppAttStatMap()).keyBy(0).timeWindow(Time.minutes(5))
			.reduce(new AppAttStatisReduce()).flatMap(new AppAttDStatisImport()).name("SDS_APP_ATT_D");
		
		//流量攻击统计天表SDS_APP_ATT_D
		winCC.union(winPort)
		.union(winIcmp)
		.union(winSQLInjection)
		.union(winXSS)
		.union(winVC)
		.union(winSynFixed)
		.union(winSynForged)
		.union(winScan)
		.union(winAck)
		.union(winBots)
		.union(winAbn)
//		.union(winAbnRe)
		.map(new FlowAttStatMap()).keyBy(0).timeWindow(Time.minutes(5))
		.reduce(new FlowAttStatisReduce()).flatMap(new FlowAttDStatisImport()).name("SDS_FLOW_ATT_D");
		

		//攻击全量统计天表SDS_ALL_ATT_D
		winCC.union(winPort)
		.union(winIcmp)
		.union(winSQLInjection)
		.union(winXSS)
		.union(winVC)
		.union(winSynFixed)
		.union(winSynForged)
		.union(winScan)
		.union(winAck)
		.union(winBots)
		.union(winAbn)
//		.union(winAbnRe)
		.map(new AllAttStatMap()).keyBy(0).timeWindow(Time.minutes(5))
		.reduce(new AllAttStatisReduce()).flatMap(new AllAttDStatisImport()).name("SDS_ALL_ATT_D");
		
	
		env.getConfig().setTaskCancellationInterval(60000);
		System.out.println("Cancel Time:" + env.getConfig().getTaskCancellationInterval());
		System.out.println("************************"); 
		System.out.println("*****planVisualizer*****");
		System.out.println("************************");
		System.out.println(env.getExecutionPlan());
		env.execute("Flink-Nyx");
	}
}
